package mytest;

import homework.WordCount;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.net.URI;

public class Test_words {

	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		// TODO Auto-generated method stub
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "Test_words");// 设置环境参数
		job.setJarByClass(WordCount.class);// 设置整个程序的类名

		job.setMapperClass(WCMapper.class);
		job.setReducerClass(WCReducer.class);

		// // reduce 输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		// 设置自定义的格式化输出
		job.setOutputFormatClass(WCFileOutputFormat.class);
		/*
		 * 分区数设置为4，因为只有四个订单，这里不需要重写分区函数，具体为什么可以看源码，
		 * 在这里大致意思就是每个key都有一个hash值,具体被分配到哪一个分区看hash%NumReduceTasks，
		 * 因为相同的Text的key相同，所以相同Text的key会被分到同一个分区。
		 */
		job.setNumReduceTasks(1);

		FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.3.101:9000/WordCount/input/test.txt"));
		// 如果文件系统已存在输出文件夹则删除
		FileSystem fs = FileSystem.get(new URI("hdfs://192.168.3.101:9000"), conf, "roger");
		if (fs.exists(new Path("/WordCount/output"))) {
			fs.delete(new Path("/WordCount/output"), true);
		}

		// 注意这里是WCFileOutPutFormat，不是FileOutPutFormat，下面同样注意
		WCFileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.3.101:9000/WordCount/output"));

		/* 消除系统的默认输出文件，不过因为文件的输出格式变了，不再是FileOutPutFormat，所以这里也要改成相对应的文件输出格式 */
		LazyOutputFormat.setOutputFormatClass(job, WCFileOutputFormat.class);

		System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}

class WCMapper extends Mapper<LongWritable, Text, Text, Text> {
	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub

		context.write(new Text(key.toString()), value);
		System.out.println(key.toString() + "@" + value.toString());
	}
}

class WCReducer extends Reducer<Text, Text, Text, Text> {

	private MultipleOutputs<Text, Text> multipleOutputs;

	@Override
	protected void setup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		multipleOutputs = new MultipleOutputs<Text, Text>(context);
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {
		String line = "";
		for (Text text : values) {
			line += text.toString();
		}

		String[] words = line.toString().split("\\s+");
		String tmp = words[0];
		for (int i = 1; i < words.length; i++) {
			tmp += "," + words[i];
		}
		// multipleOutputs.write(Key输出，value输出，输出的文件名)
		multipleOutputs.write(new Text(tmp), new Text(""), "result");
		System.out.println(tmp.toString());

	}

	@Override
	protected void cleanup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub

		multipleOutputs.close();
	}
}

class WCFileOutputFormat extends TextOutputFormat<Text, Text> {
	@Override
	public Path getDefaultWorkFile(TaskAttemptContext context, String extension) throws IOException {
		// TODO Auto-generated method stub
		FileOutputCommitter fileOutputCommitter = (FileOutputCommitter) getOutputCommitter(context);
		return new Path(fileOutputCommitter.getWorkPath(), getOutputName(context));
	}
}